package com.atguigu.loginfail_detect

import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

/**
  * @author: yangShen
  * @Description: 恶意登录监控： 如果一个用户短时间内频繁登录失败，就有可能是出现了程序的恶意攻击，比如密码暴力破解
  *              需求：在固定时间2秒内，如果登录失败达到2次则认为在恶意攻击
  *
  *              该实现存在的问题：定时器是在2秒后触发的，触发时检查登录的错误次数3是否超过设置值2 答案是超过，
  *                       但2秒内第四次为success，会清空状态不会报警
  * @Date: 2020/5/7 15:26 
  */
//输入的登录事件样例类
case class LoginEvent(userId: Long, ip: String, eventType: String, eventTime: Long)
//输出的异常报警样例类
case class Warning(userId: Long, firstFailTime: Long, lastFailTime: Long, warningMsg: String)

object LoginFail {
  def main(args: Array[String]): Unit = {
    val environment = StreamExecutionEnvironment.getExecutionEnvironment
    environment.setParallelism(1)
    environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    val source = getClass.getResource("/LoginLog.csv")
    val loginEventStream = environment.readTextFile(source.getPath)
      .map(data => {
        val dataArray = data.split(",")
        LoginEvent(dataArray(0).trim.toLong, dataArray(1).trim, dataArray(2).trim, dataArray(3).trim.toLong)
      })
      //设置时间戳
      //方式四：处理乱序数据，水位线waterMark设置为5毫秒，延时5毫秒上涨水位
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[LoginEvent](Time.seconds(5)) {
        override def extractTimestamp(element: LoginEvent): Long = element.eventTime * 1000L
      })

    val warningStream = loginEventStream
      .keyBy(_.userId)    //以用户ID进行分组
      //底层api，放大招
      .process(new LoginWarningOptimize(2))

    warningStream.print()

    environment.execute("login fail job")
  }
}

 class LoginWarning(maxFailTimes: Int) extends KeyedProcessFunction[Long, LoginEvent, Warning]{

  //定义状态，保存2秒内的所有登录失败事件
  lazy val loginFailState: ListState[LoginEvent] = getRuntimeContext.getListState(new ListStateDescriptor[LoginEvent]("login-fail-state", classOf[LoginEvent]))

  override def processElement(value: LoginEvent, ctx: KeyedProcessFunction[Long, LoginEvent, Warning]#Context, out: Collector[Warning]): Unit = {

    val loginFailList = loginFailState.get()

    //判断类型是否是fail，只添加fail的事件到状态
    if (value.eventType == "fail"){
      //如果存在登录失败的事件，则注册2秒的定时器
      if (loginFailList.iterator().hasNext){
        ctx.timerService().registerEventTimeTimer(value.eventTime * 1000L + 2000L)
      }
      loginFailState.add(value)
    }else {
      //如果在触发定时器之前成功了，则清空状态
      loginFailState.clear()
    }
  }

  //定时器回调函数
  override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[Long, LoginEvent, Warning]#OnTimerContext, out: Collector[Warning]): Unit = {

    val allLoginFails: ListBuffer[LoginEvent] = new ListBuffer[LoginEvent]()

    //触发定时器的时候，根据状态里的失败个数来决定是否输出报警
    val iter = loginFailState.get().iterator()
    while (iter.hasNext){
      allLoginFails += iter.next()
    }

    //判断个数
    if (allLoginFails.length >= maxFailTimes){
      out.collect(Warning(allLoginFails.head.userId, allLoginFails.head.eventTime, allLoginFails.last.eventTime, "login fail in 2 seconds for " + allLoginFails.length + "times."))
    }

    //触发器触发后，清空状态
    loginFailState.clear()
  }
}